package com.atguigu.pro

import java.lang
import java.net.URL

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

/**
 *
 * @description: 统计页面uv,针对一亿条数据通过布隆过滤器和redis进行优化
 * @time: 2021-03-30 21:48
 * @author: baojinlong
 * */


object UniqueVisitor2 {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 方便测试全局并行度为1
    env.setParallelism(1)

    val resource: URL = getClass.getResource("/UserBehavior-short.csv")
    // 从文件中读取数据
    // val inputStream: DataStream[String] = env.readTextFile(resource.getPath)
    // C:/codes/scala/FlinkTutorial/src/main/resources/UserBehavior.csv  E:/big-data/FlinkTutorial/src/main/resources/UserBehavior.csv
    val inputStream: DataStream[String] = env.readTextFile("C:/codes/scala/FlinkTutorial/src/main/resources/UserBehavior.csv")

    // 转换成样例类类型并提取时间戳和watermark
    val dataStream: DataStream[UserBehavior2] =
      inputStream
        .map(data => {
          val dataArray: Array[String] = data.split(",")
          UserBehavior2(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
        })
        .assignAscendingTimestamps(_.timestamp * 1000)

    val result: DataStream[UvCount] = dataStream
      .filter(x => "pv".equals(x.behavior))
      .map(data => ("uv", data.userId))
      .keyBy(_._1)
      // 每个WindowAssigner都带有一个默认触发器
      .timeWindow(Time.hours(1))
      .trigger(new MyUvTrigger) // 自定义触发器,既触发计算又清空状态
      .process(new UvCountWithBloom) // 要么增量聚合函数要么全窗口函数
    result.print("result")
    env.execute("pvJobTest")

  }
}

//class UvCountWithBloom extends KeyWindowFunction

// 触发器,每次来一条数据,直接触发窗口计算并清空窗口状态,当前数据类型,当前window类型,触发器文档:https://blog.csdn.net/lp284558195/article/details/103802738
class MyUvTrigger extends Trigger[(String, Long), TimeWindow] {
  // 每次事件回调(每来数据的时候)进入窗口的每个元素都会调用该方法 FIRE_AND_PURE:触发计算和清除窗口元素
  override def onElement(t: (String, Long), l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.FIRE_AND_PURGE

  // 系统时间改变回调 处理时间timer触发的时候会被调用 CONTINUE:什么都不做
  override def onProcessingTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  // watermark回调 事件时间timer触发的时候被调用 FIRE:触发计算
  override def onEventTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  // 收尾工作 执行窗口的删除操作 PURE:清除窗口的元素
  override def clear(w: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {}
}


// 自定义一个布隆过滤器,主要就是位图和hash函数
class MyBloom(size: Long) extends Serializable {
  /**
   * 内存容量大小
   */
  val cap: Long = size

  // hash函数
  def hash(value: String, seed: Int): Long = {
    var result = 0
    for (i <- 0 until value.length) {
      result = result * seed + value.charAt(i)
    }
    // 返回hash值,要映射到cap范围内
    (cap - 1) & result
  }
}

// 实现自定义的窗口处理函数ProcessWindowFunction全窗口函数
class UvCountWithBloom extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow] {
  lazy val jedis = new Jedis("localhost", 6379)
  lazy val bloomFilter = new MyBloom(1 << 29) // 位的个数:2^6(64)* 2^20 (1M) * 2^3(bit),64MB
  // 全窗口函数一般是窗口触发后进行调用,但是有了trigger后相当于一条数据调用一次.本来是收集齐所有数据,窗口触发计算的时候才会调用,现在是每来一条数据就调用一次
  override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
    // 先定义redis中存储位图的key
    val storedBitMapKey: String = context.window.getEnd.toString
    // 另外将当前窗口的uv count值作为状态保存到redis里,用一个叫uvcount的hash表来保存(windowEnd,count)
    val uvCountMap = "uvcount"
    val currentKey: String = storedBitMapKey
    var count: Long = 0
    // 从redis中取出当前窗口的uv count值
    val redisValue: String = jedis.hget(uvCountMap, currentKey)
    if (redisValue != null) {
      count = redisValue.toLong
    }
    // 去重:判断当前userId的hash值对应的位图位置,是否为0
    val userId: String = elements.last._2.toString
    // 计算hash值,就对应这位图中的偏移量,seed是随机的,随便写
    val offset: Long = bloomFilter.hash(userId, 61)
    // 用redis的位操作命令,取bitmap中对应位的值
    val isExist: lang.Boolean = jedis.getbit(storedBitMapKey, offset)
    if (!isExist) {
      // 如果不存在那么位图对应位置置为1,并且将count值+1
      jedis.setbit(storedBitMapKey, offset, true)
      jedis.hset(uvCountMap, currentKey, (count + 1).toString)
    }
  }
}